package com.bw.utils;

import com.bw.bean.CustomerConsumption;
import com.bw.bean.ShopCustomerMetric;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.HashSet;
import java.util.Set;

public class CustomerMetricService {

    /**
     * 聚合函数：计算店铺客户指标
     */
    public static class CustomerMetricAggregate implements AggregateFunction<CustomerConsumption, Tuple2<Set<Long>, Double>, Tuple2<Long, Double>> {

        @Override
        public Tuple2<Set<Long>, Double> createAccumulator() {
            return new Tuple2<>(new HashSet<>(), 0.0);
        }

        @Override
        public Tuple2<Set<Long>, Double> add(CustomerConsumption consumption, Tuple2<Set<Long>, Double> accumulator) {
            accumulator.f0.add(consumption.getCustomerId());
            accumulator.f1 += consumption.getAmount();
            return accumulator;
        }

        @Override
        public Tuple2<Long, Double> getResult(Tuple2<Set<Long>, Double> accumulator) {
            return new Tuple2<>( (long) accumulator.f0.size(), accumulator.f1 );
        }

        @Override
        public Tuple2<Set<Long>, Double> merge(Tuple2<Set<Long>, Double> a, Tuple2<Set<Long>, Double> b) {
            a.f0.addAll(b.f0);
            a.f1 += b.f1;
            return a;
        }
    }

    /**
     * 窗口处理函数：计算并输出店铺客户指标
     */
    public static class CustomerMetricProcessWindow extends ProcessWindowFunction<
            Tuple2<Long, Double>, ShopCustomerMetric, Long, TimeWindow> {

        private ValueState<Long> previousCustomerCount;

        @Override
        public void open(Configuration parameters) throws Exception {
            previousCustomerCount = getRuntimeContext().getState(
                    new ValueStateDescriptor<>("previousCustomerCount", Long.class, 0L)
            );
        }

        @Override
        public void process(Long shopId, Context context, Iterable<Tuple2<Long, Double>> elements, Collector<ShopCustomerMetric> out) throws Exception {
            Tuple2<Long, Double> result = elements.iterator().next();
            Long customerCount = result.f0;
            Double totalConsumption = result.f1;

            // 计算新增客户数
            Long prevCount = previousCustomerCount.value();
            Long newCustomer = Math.max(0, customerCount - prevCount);

            // 计算平均消费金额
            Double avgConsumption = customerCount > 0 ? totalConsumption / customerCount : 0;

            // 计算客户留存率 (简化计算：假设为80%)
            Double retentionRate = 0.8;

            // 构建指标对象
            ShopCustomerMetric metric = new ShopCustomerMetric();
            metric.setShopId(shopId);
            metric.setStatTime(new Timestamp(context.window().getEnd()));
            metric.setTotalCustomer(customerCount);
            metric.setNewCustomer(newCustomer);
            metric.setTotalConsumption(totalConsumption);
            metric.setAvgConsumption(avgConsumption);
            metric.setCustomerRetentionRate(retentionRate);
            metric.setActiveCustomer(customerCount); // 简化处理，活跃客户数=总客户数

            // 更新历史客户数
            previousCustomerCount.update(customerCount);

            out.collect(metric);
        }
    }
}
